package com.szsh.bigdata.open.utils

import org.apache.commons.lang.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.IOUtils

import java.io.{File, FileInputStream, FileOutputStream, IOException}

object HdfsUtils {
	

	private[this] val ARGS = Array(("fs.defaultFS", "hdfs://xxx.xxx.xxx.xxx:port/"))
	
	/**
	 * 获取fs操作对象
	 *
	 * @return fs
	 */
	private[this] def getFS: FileSystem = {
		val conf = new Configuration()
		for (args <- ARGS) conf.set(args._1, args._2)
		FileSystem.get(conf)
	}
	
	/**
	 * 关闭FileSystem
	 *
	 * @param fs fs
	 */
	private[this] def closeFS(fs: FileSystem): Unit = {
		if (fs != null) {
			try {
				fs.close()
			} catch {
				case e: IOException => e.printStackTrace()
			}
		}
	}
	
	
	def getFilesDetail(hdfsFilePath: String, fileSystem: FileSystem = null): List[Path] = {
		var list: List[Path] = List()
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val fsStats = fs.listStatus(new Path(hdfsFilePath))
		for (fstat <- fsStats) {
			println(s"------------------$fstat------------------")
			println("Permission:" + fstat.getPermission)
			println("Owner:" + fstat.getOwner)
			println("Group:" + fstat.getGroup)
			println("Size:" + fstat.getLen)
			println("Replication:" + fstat.getReplication)
			println("Block Size:" + fstat.getBlockSize)
			println("Name:" + fstat.getPath)
			list = list :+ fstat.getPath
		}
		if (fileSystem == null) closeFS(fs)
		list
	}
	
	
	/**
	 * 列出目录下文件细节
	 *
	 * @param hdfsFilePath 文件目录
	 */
	def listFilesDetail(hdfsFilePath: String, fileSystem: FileSystem = null): Unit = {
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val fsStats = fs.listStatus(new Path(hdfsFilePath))
		for (fstat <- fsStats) {
			println(s"------------------$fstat------------------")
			println("Permission:" + fstat.getPermission)
			println("Owner:" + fstat.getOwner)
			println("Group:" + fstat.getGroup)
			println("Size:" + fstat.getLen)
			println("Replication:" + fstat.getReplication)
			println("Block Size:" + fstat.getBlockSize)
			println("Name:" + fstat.getPath)
		}
		if (fileSystem == null) closeFS(fs)
	}
	
	
	/**
	 * 创建目录
	 *
	 * @param hdfsFilePath 目录
	 */
	def mkdir(hdfsFilePath: String, fileSystem: FileSystem = null): Unit = {
		val fs = if (fileSystem != null) fileSystem else this.getFS
		try {
			val success = fs.mkdirs(new Path(hdfsFilePath))
			if (success) {
				println("Create directory or file successfully")
			}
		} catch {
			case e: IllegalArgumentException => e.printStackTrace()
			case e: IOException => e.printStackTrace()
		} finally {
			if (fileSystem == null) closeFS(fs)
		}
	}
	
	/**
	 * 删除文件或目录
	 *
	 * @param hdfsFilePath 文件或目录
	 * @param recursive    递归
	 */
	def deleteHdfs(hdfsFilePath: String, recursive: Boolean, fileSystem: FileSystem = null): Unit = {
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val path = new Path(hdfsFilePath)
		try {
			if (fs.exists(path)) {
				val success = fs.delete(path, recursive)
				if (success) {
					System.out.println("delete successfully")
				}
			}
		} catch {
			case e: IllegalArgumentException => e.printStackTrace()
			case e: IOException => e.printStackTrace()
		} finally {
			if (fileSystem == null) closeFS(fs)
		}
	}
	
	
	/**
	 * 上传文件到HDFS
	 *
	 * @param localPath  localPath
	 * @param hdfsPath   hdfsPath
	 * @param fileSystem fs
	 */
	def write(localPath: String, hdfsPath: String, fileSystem: FileSystem = null): Unit = {
		
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val inStream = new FileInputStream(new File(localPath))
		val outStream = fs.create(new Path(hdfsPath))
		try {
			IOUtils.copyBytes(inStream, outStream, 4096, false)
		} catch {
			case e: IOException => e.printStackTrace()
		} finally {
			IOUtils.closeStream(inStream)
			IOUtils.closeStream(outStream)
			if (fileSystem == null) this.closeFS(fs)
		}
		
	}

	/**
	 * 判断路径是否存在
	 * @param path hdfs路径
	 * @return
	 */
	def isExist(path: String, fileSystem: FileSystem = null): Boolean = {
		var result = false
		val fs = if (fileSystem != null) fileSystem else this.getFS
		try {
			if (fs.exists(new Path(path))) {
				result = true
			}
		}catch {
			case ex: Exception => println(ex)
		}finally {
			if (fileSystem == null) this.closeFS(fs)
		}
		result

	}
	
	
	/**
	 * 打印hdfs上的文件内容
	 *
	 * @param hdfsFilePath 文件目录
	 * @param fileSystem   fs
	 */
	def cat(hdfsFilePath: String, fileSystem: FileSystem = null): Unit = {
		
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val inStream = fs.open(new Path(hdfsFilePath))
		try {
			IOUtils.copyBytes(inStream, System.out, 4096, false)
		} catch {
			case e: IOException => e.printStackTrace()
		} finally {
			IOUtils.closeStream(inStream)
			if (fileSystem == null) this.closeFS(fs)
		}
	}
	
	/**
	 * 下载文件到本地
	 *
	 * @param localFilePath 本地路径
	 * @param hdfsFilePath  hdfs路径
	 */
	def getMerge(localFilePath: String, hdfsFilePath: String, fileSystem: FileSystem = null): Unit = {
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val fsStats = fs.listStatus(new Path(hdfsFilePath))
		val fos = new FileOutputStream(new File(localFilePath))
		try {
			for (fsStat <- fsStats) {
				val filePath = fsStat.getPath.toString
				val fsIns = fs.open(new Path(filePath))
				try {
					println(s"正在合并小文件${filePath}")
					IOUtils.copyBytes(fsIns, fos, 1024)
				}catch {
					case e: Exception => e.printStackTrace()
				}finally {
					IOUtils.closeStream(fsIns)
				}
			}
		} catch {
			case e: IllegalArgumentException => e.printStackTrace()
			case e: IOException => e.printStackTrace()
		} finally {
			if (fileSystem == null) this.closeFS(fs)
		}
	}
	
	
	/**
	 * 下载文件到本地
	 *
	 * @param localFilePath 本地路径
	 * @param hdfsFilePath  hdfs路径
	 */
	def get(localFilePath: String, hdfsFilePath: String, fileSystem: FileSystem = null): Unit = {
		val fs = if (fileSystem != null) fileSystem else this.getFS
		val fsIs = fs.open(new Path(hdfsFilePath))
		val fos = new FileOutputStream(new File(localFilePath))
		try {
			IOUtils.copyBytes(fsIs, fos, 1024)
		} catch {
			case e: IllegalArgumentException => e.printStackTrace()
			case e: IOException => e.printStackTrace()
		} finally {
			IOUtils.closeStream(fsIs)
			if (fileSystem == null) this.closeFS(fs)
		}
	}

	/**
	 * 上传本地文件到HDFS
	 * @param localPath      本地文件路径
	 * @param hdfsPath       HDFS文件路径
	 * @param isDelLocalFile 是否删除本地文件
	 * @return
	 */
	@deprecated
	def push(localPath: String, hdfsPath: String, isDelLocalFile: Boolean): Boolean = {
		var result = false
		val hdfs = getFS
		try {
			if (StringUtils.isNotBlank(localPath) && StringUtils.isNotBlank(hdfsPath)) {
				val realUrl = "hdfs://" + hdfsPath
				val src = new Path(localPath)
				val dst = new Path(realUrl)
				if (isDelLocalFile) {
					hdfs.copyFromLocalFile(true, true, src, dst)
				} else {
					hdfs.copyFromLocalFile(false, src, dst)
				}
				result = true
			}
		} catch {
			case e: Exception => println(e)
		}
		result
	}


}